package com.qy566.gps.pulsar.demo;

import org.apache.pulsar.client.api.*;

import java.util.concurrent.atomic.LongAdder;

/**
 * @author wangnaiqi
 * @description TODO
 * @date 2023/11/8
 */
public class ConsumerDemo {
    public static LongAdder adder = new LongAdder();
    public static void main(String[] args) throws PulsarClientException {
        PulsarClient client = PulsarClient.builder()
                .serviceUrl("pulsar://10.0.0.27:30980")
                .authentication(AuthenticationFactory.token("eyJhbGciOiJSUzI1NiJ9.eyJzdWIiOiJhZG1pbiJ9.UUlFnlvhcPHMWPNAebzWwt6R71MxKgUIKZEZA3WeJ7d74AY2O5vFit6CcF4QyMyJri8EAzLDidkrkxmRwJ0egzqmEZLYDtKtAwBSKBrucoOzsZ-gTu_escnt_Qtzpa7QGcE2KX7wpW3-2BuolBrnmaipQqoZrq54Zlse2qA9bJgObJmXAMvIAO6tmZEHqtsWCzZV1jfXjiHE8SlESeVQ8TpvoyUzD6R2Bx36LEGRACNYxs_k0E2FO0cSflWB3sCS7mOE8ksqstNLdiN54ECfIk3AlQ0ahkvRk8b4Gweq0Vv2E87NBIFVYpYKyDFiUTlVcRaCr0RQk96J6T2qVsoTfg"))
                .listenerThreads(200)
                .build();
        MessageListener myMessageListener = new MessageListener() {
            @Override
            public void received(Consumer consumer, Message msg) {
                try {
                    System.out.println("收到消息: " + new String(msg.getData())+"当前条数"+adder.toString());
                    adder.add(1);
                    consumer.acknowledge(msg);
                } catch (Exception e) {
                    consumer.negativeAcknowledge(msg);
                }
            }
        };
        Consumer consumer = client.newConsumer(Schema.STRING)
                .topic("persistent://gpsTenant/positionNamespace/positionTopic")
                .subscriptionName("devicePositionMsgTopicSubscription")
                .subscriptionType(SubscriptionType.Shared)
                .messageListener(myMessageListener)
                .subscribe();
    }
}
